package com.wudl.szse;

import com.wudl.avro.SzseAvro;
import com.wudl.kafka.KafkaPro;

import java.io.DataInputStream;
import java.io.InputStream;
import java.net.Socket;

/**
 * @author ：wudl
 * @date ：Created in 2021-12-17 0:31
 * @description：
 * @modified By：
 * @version: 1.0
 */

public class SocketClient2 {


    //1.创建main方法
    public static void main(String[] args) throws Exception {

        /**
         * 开发步骤:
         * 1.创建main方法
         * 2.建立socket连接，获取流数据
         * 3.解析行数据，数据转换
         * 4.发送kafka
         */
        // 2.建立socket连接，获取流数据
        Socket socket = new Socket("localhost", 4444);

        //读取数据
        InputStream inputStream = socket.getInputStream();
        DataInputStream dataInputStream = new DataInputStream(inputStream);

        //新建kafka生产者对象
        KafkaPro kafkaPro = new KafkaPro();
        while (true) {

            //读取每一条数据
            String str = dataInputStream.readUTF();
            //4.解析行数据，数据转换
            SzseAvro szseAvro = parseStr(str);
            //5.发送kafka
            kafkaPro.sendData("szse",szseAvro);
            System.out.println(">>>>>>>:"+szseAvro);
        }

    }

    //解析字符串，转换成avro对象
    private static SzseAvro parseStr(String str) {

        //1.字符串切割,获取各字段数据
        String[] arr = str.split("\\|");

        //8.封装结果数据
        SzseAvro szseAvro = new SzseAvro();
        szseAvro.setMdStreamID(arr[0]);
        szseAvro.setSecurityID(arr[1]);
        szseAvro.setSymbol(arr[2]);
        szseAvro.setTradeVolume(Long.valueOf(arr[3])); //成交总量
        szseAvro.setTotalValueTraded(Long.valueOf(arr[4]));//成交总额
        szseAvro.setPreClosePx(new Double(arr[5].trim()));
        szseAvro.setOpenPrice(new Double(arr[6].trim()));
        szseAvro.setHighPrice(new Double(arr[7].trim()));
        szseAvro.setLowPrice(new Double(arr[8].trim()));
        szseAvro.setTradePrice(new Double(arr[9].trim()));
        szseAvro.setClosePx(new Double(arr[10].trim())); //在15点之前，收盘价和最新家数据一样
        szseAvro.setTradingPhaseCode(arr[11]); //处于连续竞价期间，市场交易未结束
        szseAvro.setTimestamp(Long.valueOf(arr[12])); //事件时间
        return szseAvro;
    }
}
